package org.example.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.example.realtime.jtp.common.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 使用状态State编程，对is_new字段值进行校验和纠偏
 * @author xuanyu
 */
public class AdjustIsNewProcessFunction extends KeyedProcessFunction<String, String, String> {

    /**
     * step1. 声明状态
     */
    private transient ValueState<String> firstDateState = null ;

    @Override
    public void open(Configuration parameters) throws Exception {
        /**
         * step2. 初始化状态
         */
        firstDateState = getRuntimeContext().getState(
                new ValueStateDescriptor<String>("firstDate", String.class)
        );
    }

    @Override
    public void processElement(String value,
                               KeyedProcessFunction<String, String, String>.Context ctx,
                               Collector<String> out) throws Exception {
        /**
         * step3. 获取状态值
         */
        String firstDate = firstDateState.value();

        // 获取ts和isNew字段值
        JSONObject jsonObject = JSON.parseObject(value);
        Long tsValue = jsonObject.getLong("ts");
        String isNewValue = jsonObject.getJSONObject("common").getString("is_new");
        /**
       firstDate = null
          isNewValue = 1 新访客
             正常，设置firstDate为当前日期
          isNewValue = 0 老访客
             正常，设置firstDate给一个日期值，比如昨天日期

       firstDate 有值情况
          isNewValue = 1 新访客
             firstDate = ts中日期，不变
             firstDate != ts中日期，纠偏修改is_new值为0
       */
//        if(null == firstDate){
//            String logDate;
//            if("1".equals(isNewValue)){
//                // 更新第一次访问日期
//                logDate = DateTimeUtil.convertLongToString(tsValue, "yyyy-MM-dd");
//            }else{
//                // 更新第一次访问日期
//                logDate = DateTimeUtil.convertLongToString(tsValue - 24 * 60 * 60 * 1000, "yyyy-MM-dd");
//            }
//            /**
//             * step4. 更新状态值
//             */
//            firstDateState.update(logDate) ;
//        }else{
//            if("1".equals(isNewValue) ||
//                    ! firstDateState.equals(DateTimeUtil.convertLongToString(tsValue, "yyyy-MM-dd")) ){
//                // 修改值
//                jsonObject.getJSONObject("common").put("is_new", "0") ;
//            }
//        }
        //todo isNew = 1 --> firstDate=null firstDate=有值
        if("1".equals(isNewValue)){
            if(null == firstDate){
                //第四步更新状态值
                //此时is_new正确，状态值更新为第一次访问日期
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue,DateTimeUtil.DATE_FORMAT));
            }else {
                //状态值（首次访问日期）等于数据日期不做处理
                if(!firstDate.equals(DateTimeUtil.convertLongToString(tsValue,DateTimeUtil.DATE_FORMAT))){
                    //此时is_new属于异常，需要纠正，设置为0
                    jsonObject.getJSONObject("common").put("is_new", "0");
                }
            }
        }
        //isNew = 0 --> firstDate=null firstDate=有值
        else{
            //当用户早早登陆访问，当时实时应用后来处理，用户第一次访问时，firstDate为null
            if(null==firstDate){
                //设置一个昨天日期即可
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue - 24 * 60 * 60 * 1000,DateTimeUtil.DATE_FORMAT));
            }
        }

        // 输出
        out.collect(jsonObject.toJSONString());
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

}